/**
 * 抖音关注：程序员三丙
 * 知识星球：https://t.zsxq.com/j9b21
 */
package sanbing.example.dylike.application;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import sanbing.example.dylike.domain.LikeCacheGateway;
import sanbing.example.dylike.infrastructure.queue.QueueMsg;
import sanbing.example.dylike.infrastructure.queue.QueueProducer;
import sanbing.example.dylike.infrastructure.queue.TopicPartitionInfo;
import sanbing.example.dylike.infrastructure.queue.provider.QueueProducerProvider;
import sanbing.example.dylike.util.JacksonUtil;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.UUID;
import java.util.concurrent.ExecutorService;


@Component
@Slf4j
public class LikeQryExe {

    @Resource
    LikeCacheGateway likeCacheGateway;

    @Resource
    QueueProducerProvider producerProvider;

    @Value("${clickhouse.like-behavioral}")
    private boolean clickhouseLikeBehavioral;

    private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    private QueueProducer clickhouseProducer;

    @PostConstruct
    public void init() {
        this.clickhouseProducer = producerProvider.getEventClickhouseMsgProducer();
    }

    @PreDestroy
    public void destroy() {
        this.clickhouseProducer.stop();
    }

    public ListenableFuture<Integer> execute(UUID vId, ExecutorService executor) {
        return Futures.transformAsync(likeCacheGateway.read(vId), input -> {

            if (!clickhouseLikeBehavioral) {
                return Futures.immediateFuture(input);
            }

            TopicPartitionInfo tpi = new TopicPartitionInfo(clickhouseProducer.getTopic(), null, false);

            UUID id = UUID.randomUUID();
            ObjectNode likeRecord = JacksonUtil.OBJECT_MAPPER.createObjectNode();
            likeRecord.put("id", id.toString());
            likeRecord.put("v_id", vId.toString());
            likeRecord.put("user_id", "2088302720161613");
            likeRecord.put("create_time", DATE_TIME_FORMATTER.format(LocalDateTime.now()));
            likeRecord.put("field_1", "field_1_value");
            likeRecord.put("field_2", "field_2_value");
            likeRecord.put("field_3", "field_3_value");
            likeRecord.put("field_4", "field_4_value");
            likeRecord.put("field_5", "field_5_value");
            likeRecord.put("field_6", "field_6_value");
            likeRecord.put("field_7", "field_7_value");
            likeRecord.put("field_8", "field_8_value");
            likeRecord.put("field_9", "field_9_value");
            likeRecord.put("field_10", "field_10_value");

            clickhouseProducer.send(tpi, new QueueMsg(id, likeRecord.toString().getBytes(StandardCharsets.UTF_8)), null);

            return Futures.immediateFuture(input);
        }, executor);
    }

}